﻿//
// © Algobox Ltd - Benn Eichhorn
//
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using Algobox.Objects.Finance.Assets.Stock;
using System.Data;
using System.Data.SqlClient;
using System.Threading;
using System.Diagnostics;
using System.Collections.Concurrent;
using Algobox.Objects.Finance;

namespace Algobox.Data.Finance.AlgoboxData
{
    public class AlgoboxDataInterface : IDisposable
    {
        public AlgoboxDataInterface(string connection)
        {
            _dbConnectionString = connection;
            _linq = new AlgoboxDataContext(connection);
        }

        public void Dispose()
        {
            _linq.Dispose();
            _continueBulkInsert = false;

            if (_bulkCopy != null)
            {
                DisposeBulkCopy();
            }
        }
        
        private static readonly NLog.Logger Log = NLog.LogManager.GetCurrentClassLogger();

        AlgoboxDataContext _linq;
        string _dbConnectionString;
        
        public bool TryGetStocksFromPricesIntraday(out List<IStock> stocks)
        {
            stocks = new List<IStock>();
            try
            {
                foreach (string symbol in _linq.zephyr_PricesIntradays.Select(a => a.Symbol).Distinct())
                {
                    stocks.Add(new StockSecurity(symbol, Objects.Finance.Markets.MarketMIC.XXXX));
                }
                return true;
            }
            catch { }
            return false;
        }


        #region Ticker Plant

        const string TABLELASTS = "Algobox.dbo.zephyr_PricesIntraday";
        const int MAX_LASTS_FLUSH = 200;
        const int MAX_BULKCOPY_FAILURES = 3;
        DataTable _dataTablePricesIntraday;
        //ConcurrentQueue<DataRow> _queueBulkUpdates;
        SqlBulkCopy _bulkCopy;
        Thread _rowDataQueueThread;
        bool _continueBulkInsert;
        long _counterBulkRowsWritten;
        long _counterTotalUpdates;
        int _flushCounter;
        DateTime _counterUpdateTimestamp;
        Stopwatch _stopwatchBulkWrites = Stopwatch.StartNew();

        public event Action<int, TimeSpan> OnBulkCollectionStatistic;
        public event Action<int, TimeSpan> OnBulkWriteStatistic;
        public event Action<string> OnBulkWriteFailure;
        public event Action<string> OnBulkCopyShutdown;

        public string FeedName { get { return "AlgoboxDataInterface"; } }

        public long BulkPriceTotalUpdates { get { return Interlocked.Read(ref _counterTotalUpdates); } }
        public long BulkPriceBulkRowsWritten { get { return _counterBulkRowsWritten; } }
        public DateTime BulkPriceUpdateTime { get { return _counterUpdateTimestamp; } }
        
        /// <summary>
        /// Initialises the DataTables for use with the bulk copy
        /// Use SqlBulkCopyColumnMapping for mapping columns in DataTable to table def in database
        /// http://msdn.microsoft.com/en-us/library/system.data.sqlclient.sqlbulkcopycolumnmapping.aspx
        /// </summary>
        private void InitialiseBulkCopy()
        {
            _continueBulkInsert = true;

            if (_bulkCopy == null)
            {
                _bulkCopy = new SqlBulkCopy(_dbConnectionString);
                _flushCounter = 0;
                _counterBulkRowsWritten = 0;
            }

            //if (_queueBulkUpdates == null)
            //{
            //    _queueBulkUpdates = new ConcurrentQueue<DataRow>();
            //}

            if (_rowDataQueueThread == null)
            {
                _rowDataQueueThread = new Thread(new ThreadStart(this.ProcessBulkCopyQueue));
                _rowDataQueueThread.Name = FeedName;
                _rowDataQueueThread.Priority = ThreadPriority.BelowNormal;
                _rowDataQueueThread.Start();
            }

            if (_dataTablePricesIntraday == null)
            {
                _dataTablePricesIntraday = new DataTable(TABLELASTS);
                DataColumn dc;
                dc = new DataColumn("UpdateDate", typeof(DateTime));
                dc.AllowDBNull = false;
                _dataTablePricesIntraday.Columns.Add(dc);
                dc = new DataColumn("UpdateTime", typeof(TimeSpan));
                dc.AllowDBNull = false;
                _dataTablePricesIntraday.Columns.Add(dc);
                dc = new DataColumn("Symbol", typeof(string));
                dc.AllowDBNull = false;
                _dataTablePricesIntraday.Columns.Add(dc);
                dc = new DataColumn("Phase", typeof(string));
                dc.AllowDBNull = false;
                _dataTablePricesIntraday.Columns.Add(dc);
                dc = new DataColumn("LastPrice", typeof(decimal));
                dc.AllowDBNull = false;
                _dataTablePricesIntraday.Columns.Add(dc);
                dc = new DataColumn("LastVolume", typeof(long));
                dc.AllowDBNull = false;
                _dataTablePricesIntraday.Columns.Add(dc);
                dc = new DataColumn("LastUpdated", typeof(bool));
                dc.AllowDBNull = false;
                _dataTablePricesIntraday.Columns.Add(dc);
                dc = new DataColumn("BidPrice", typeof(decimal));
                dc.AllowDBNull = false;
                _dataTablePricesIntraday.Columns.Add(dc);
                dc = new DataColumn("BidVolume", typeof(long));
                dc.AllowDBNull = false;
                _dataTablePricesIntraday.Columns.Add(dc);
                dc = new DataColumn("BidUpdated", typeof(bool));
                dc.AllowDBNull = false;
                _dataTablePricesIntraday.Columns.Add(dc);
                dc = new DataColumn("AskPrice", typeof(decimal));
                dc.AllowDBNull = false;
                _dataTablePricesIntraday.Columns.Add(dc);
                dc = new DataColumn("AskVolume", typeof(long));
                dc.AllowDBNull = false;
                _dataTablePricesIntraday.Columns.Add(dc);
                dc = new DataColumn("AskUpdated", typeof(bool));
                dc.AllowDBNull = false;
                _dataTablePricesIntraday.Columns.Add(dc);
            }
        }

        private void DisposeBulkCopy()
        {
            DataRow row;
            long elapsed = 1;

            Log.Info("DisposeBulkCopy saving remaining data");


            _stopwatchBulkWrites.Stop();
            elapsed = _stopwatchBulkWrites.ElapsedMilliseconds == 0 ? 1 : _stopwatchBulkWrites.ElapsedMilliseconds;
            _stopwatchBulkWrites.Reset();
            
            lock (_dataTablePricesIntraday)
            {
                if (Log.IsInfoEnabled)
                {
                    Log.Info("{0} price updates collected in {1:0.00} seconds",
                        _dataTablePricesIntraday.Rows.Count,
                        elapsed / 1000.0);
                }
                if (OnBulkCollectionStatistic != null)
                {
                    OnBulkCollectionStatistic(_dataTablePricesIntraday.Rows.Count, TimeSpan.FromMilliseconds(elapsed));
                }

                _bulkCopy.DestinationTableName = TABLELASTS;
                try
                {
                    _stopwatchBulkWrites.Start();
                    // bulk copy the data to the database
#if !DEBUG
                    _bulkCopy.WriteToServer(_dataTablePricesIntraday);
#endif
                    _stopwatchBulkWrites.Stop();
                    elapsed = _stopwatchBulkWrites.ElapsedMilliseconds == 0 ? 1 : _stopwatchBulkWrites.ElapsedMilliseconds;

                    if (Log.IsInfoEnabled)
                    {
                        Log.Info("{0} price updates saved to database in {1} milliseconds",
                            _dataTablePricesIntraday.Rows.Count,
                            elapsed);
                    }

                    if (OnBulkWriteStatistic != null)
                    {
                        OnBulkWriteStatistic(_dataTablePricesIntraday.Rows.Count, TimeSpan.FromMilliseconds(elapsed));
                    }

                    _counterBulkRowsWritten += _dataTablePricesIntraday.Rows.Count;
                    _dataTablePricesIntraday.Rows.Clear();
                }
                catch (Exception e)
                {
                    Log.Error("Failed to bulk write price updates to server, error message was: " + e.Message);

                    string message = String.Format("{0:O} {1}", DateTime.Now, e.Message);

                    if (OnBulkWriteFailure != null)
                        OnBulkWriteFailure(message);
                }
                _stopwatchBulkWrites.Reset();
            }
        }

        public void QueueQuoteUpdate(DateTime timeStamp, string symbol, Phases.Phase phase, double lastPrice, int lastSize, bool lastUpdated, double bidPrice, int bidSize, bool bidUpdated, double askPrice, int askSize, bool askUpdated)
        {
            if (!_continueBulkInsert)
            {
                if (_bulkCopy == null)
                {
                    InitialiseBulkCopy();
                }
                else
                {
                    return;
                }
            }

            Interlocked.Increment(ref _counterTotalUpdates);
            _counterUpdateTimestamp = timeStamp;

            try
            {
                DataRow rowLast = _dataTablePricesIntraday.NewRow();
                rowLast["UpdateDate"] = timeStamp.Date;
                rowLast["UpdateTime"] = timeStamp.TimeOfDay;
                rowLast["Symbol"] = symbol;
                rowLast["Phase"] = phase.ToString();
                rowLast["LastPrice"] = (decimal)lastPrice;
                rowLast["LastVolume"] = lastSize;
                rowLast["LastUpdated"] = lastUpdated;
                rowLast["BidPrice"] = (decimal)bidPrice;
                rowLast["BidVolume"] = bidSize;
                rowLast["BidUpdated"] = bidUpdated;
                rowLast["AskPrice"] = (decimal)askPrice;
                rowLast["AskVolume"] = askSize;
                rowLast["AskUpdated"] = askUpdated;

                lock(_dataTablePricesIntraday)
                    _dataTablePricesIntraday.Rows.Add(rowLast); 
            }
            catch (Exception e)
            {
                Log.Debug("QueueQuoteUpdate Lost data point due to " + e.Message);
            }
     
        }

        private void ProcessBulkCopyQueue()
        {
            // second delay before saving data when there is no updates
            int counterMAX = 30;
            int countLast = 0;
            int counterLast = 0;
            int failures = 0;
            long elapsed = 1;
            
            DataTable table;

            while (_continueBulkInsert)
            {
                if (_dataTablePricesIntraday.Rows.Count > MAX_LASTS_FLUSH
                    || counterLast >= counterMAX
                    || _flushCounter > 0)
                {
                    _stopwatchBulkWrites.Stop();
                    elapsed = _stopwatchBulkWrites.ElapsedMilliseconds == 0 ? 1 : _stopwatchBulkWrites.ElapsedMilliseconds;
                    _stopwatchBulkWrites.Reset();

                    lock (_dataTablePricesIntraday)
                    {
                        table = _dataTablePricesIntraday.Copy();
                        _dataTablePricesIntraday.Rows.Clear();
                    }

                    if (Log.IsInfoEnabled)
                    {
                        Log.Info("{0} Price updates collected in {1:0.00} seconds",
                            table.Rows.Count,
                            elapsed / 1000.0);
                    }
                    if (OnBulkCollectionStatistic != null)
                    {
                        OnBulkCollectionStatistic(table.Rows.Count, TimeSpan.FromMilliseconds(elapsed));
                    }

                    _bulkCopy.DestinationTableName = TABLELASTS;
                    try
                    {
                        _stopwatchBulkWrites.Start();
                        // bulk copy the data to the database
#if !DEBUG
                        _bulkCopy.WriteToServer(table);
#endif
                        _stopwatchBulkWrites.Stop();
                        elapsed = _stopwatchBulkWrites.ElapsedMilliseconds == 0 ? 1 : _stopwatchBulkWrites.ElapsedMilliseconds;

                        if (Log.IsInfoEnabled)
                        {
                            Log.Info("{0} Price updates saved to database in {1} milliseconds",
                                table.Rows.Count,
                                elapsed);
                        }

                        if (OnBulkWriteStatistic != null)
                        {
                            OnBulkWriteStatistic(table.Rows.Count, TimeSpan.FromMilliseconds(elapsed));
                        }

                        _counterBulkRowsWritten += table.Rows.Count;
                        failures = 0;
                    }
                    catch (Exception e)
                    {
                        Log.Error("Failed to bulk write price updates to server, error message was: " + e.ToString());

                        string message = String.Format("{0:O} Fail {1} of {2} for {3}", DateTime.Now,failures,MAX_BULKCOPY_FAILURES, e.Message);

                        if (++failures >= MAX_BULKCOPY_FAILURES)
                        {
                            Log.Error("To many attempts to bulk copy failed so stopping bulk copy");
                            _continueBulkInsert = false;
                            
                            if (OnBulkWriteFailure != null)
                                OnBulkWriteFailure(message);

                            if(OnBulkCopyShutdown != null)
                                OnBulkCopyShutdown(message);

                            break;
                        }

                        if (OnBulkWriteFailure != null)
                            OnBulkWriteFailure(message);

                        lock(_dataTablePricesIntraday)
                            _dataTablePricesIntraday.Merge(table);
                    }
                    table.Clear();
                    _stopwatchBulkWrites.Reset();
                    _stopwatchBulkWrites.Start();
                    countLast = 0;
                    counterLast = 0;
                }
                else if (_dataTablePricesIntraday.Rows.Count > 0)
                {
                    if (_dataTablePricesIntraday.Rows.Count == countLast)
                    {
                        ++counterLast;
                    }
                    else
                    {
                        counterLast = 0;
                        countLast = _dataTablePricesIntraday.Rows.Count;
                    }
                }

                if (_flushCounter > 0)
                    --_flushCounter;

                // sleep if there is nothing to do
                if (_flushCounter == 0)
                    Thread.Sleep(1000);
            }
        }


        #endregion


    }
}
